package cn.tedu.stream.latedata

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * @author Amos
 * @date 2022/5/24
 */

object StreamLateDataDemo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 指定使用eventTime处理数据
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 构建数据源 timestamp hello
    val source = env.socketTextStream("hadoop01", 9999)
    // 抽取eventTime的字段并且指定watermark
    val waterMarkStream: DataStream[String] = source.assignTimestampsAndWatermarks(
      // 构造参数中，指定watermark的时间
      new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(2)) {
        /**
         * 指定时间戳字段
         *
         * @param element  : 记录的一条数据
         * @return
         */
        override def extractTimestamp(element: String): Long = {
          val eventTime = element.split(" ")(0).toLong
          println(eventTime)
          eventTime
        }
      }
    )

    // timestamp hello => (hello,1)
    import org.apache.flink.api.scala._
    val keyedStream: KeyedStream[(String, Int), Tuple] = waterMarkStream.map(x => (x.split(" ")(1), 1)).keyBy(0)

    // 定义变量，接收侧输出流
    val lateData = new OutputTag[(String, Int)]("late")

    // 指定窗口eventTime的滚动窗口
    val windowStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)))
      .allowedLateness(Time.seconds(2)) // 对于再次迟到数据的处理
      .sideOutputLateData(lateData) // 将再次迟到的数据放入到侧输出流的数据中

    // 数据的聚合
    val result = windowStream.sum(1)

    result.print()
    // 获取最后迟到且没有处理的数据
    val late: DataStream[(String, Int)] = result.getSideOutput(lateData)
    late.print("late")

    env.execute()
  }

}
